"""
熔断器模块 - 防止级联故障

这个模块实现了熔断器模式，用于防止级联故障：
- 监控服务调用成功率
- 自动熔断和恢复
- 支持多种熔断策略
- 提供降级处理机制
"""

import asyncio
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from enum import Enum
from typing import Any, Callable, Dict, List, Optional, Union
import statistics

logger = logging.getLogger(__name__)


class CircuitState(Enum):
    """熔断器状态"""
    CLOSED = "closed"      # 正常状态，允许请求通过
    OPEN = "open"          # 熔断状态，拒绝所有请求
    HALF_OPEN = "half_open"  # 半开状态，允许部分请求测试


class FailureType(Enum):
    """失败类型"""
    TIMEOUT = "timeout"
    EXCEPTION = "exception"
    HTTP_ERROR = "http_error"
    CUSTOM = "custom"


@dataclass
class CircuitBreakerConfig:
    """熔断器配置"""
    # 熔断阈值
    failure_threshold: int = 5  # 连续失败次数阈值
    failure_rate_threshold: float = 0.5  # 失败率阈值
    success_threshold: int = 3  # 半开状态下成功次数阈值
    
    # 时间窗口
    timeout_window: int = 60  # 时间窗口（秒）
    recovery_timeout: int = 30  # 恢复超时时间（秒）
    
    # 请求超时
    request_timeout: float = 10.0  # 单个请求超时时间（秒）
    
    # 监控配置
    min_requests: int = 10  # 最小请求数（用于计算失败率）
    max_requests: int = 1000  # 最大请求数（滑动窗口大小）
    
    # 降级配置
    enable_fallback: bool = True  # 是否启用降级
    fallback_delay: float = 0.1  # 降级响应延迟（秒）


@dataclass
class RequestRecord:
    """请求记录"""
    timestamp: float
    success: bool
    duration: float
    failure_type: Optional[FailureType] = None
    error_message: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class CircuitBreakerStats:
    """熔断器统计信息"""
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    timeout_requests: int = 0
    circuit_opens: int = 0
    circuit_closes: int = 0
    fallback_calls: int = 0
    current_state: CircuitState = CircuitState.CLOSED
    last_state_change: Optional[datetime] = None
    failure_rate: float = 0.0
    average_response_time: float = 0.0


class CircuitBreaker:
    """熔断器实现"""
    
    def __init__(self, 
                 name: str,
                 config: Optional[CircuitBreakerConfig] = None,
                 fallback_func: Optional[Callable] = None):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        self.fallback_func = fallback_func
        
        # 状态管理
        self.state = CircuitState.CLOSED
        self.last_failure_time: Optional[float] = None
        self.consecutive_failures = 0
        self.consecutive_successes = 0
        
        # 请求记录
        self.request_records: List[RequestRecord] = []
        self.lock = asyncio.Lock()
        
        # 统计信息
        self.stats = CircuitBreakerStats()
        
        self.logger = logging.getLogger(f"{__name__}.{name}")
        self.logger.info(f"熔断器 {name} 已初始化")
    
    async def call(self, 
                   func: Callable,
                   *args,
                   **kwargs) -> Any:
        """执行函数调用，应用熔断器逻辑"""
        async with self.lock:
            # 检查熔断器状态
            if not await self._should_allow_request():
                self.stats.fallback_calls += 1
                return await self._handle_fallback(*args, **kwargs)
            
            # 执行请求
            start_time = time.time()
            try:
                # 设置超时
                if asyncio.iscoroutinefunction(func):
                    result = await asyncio.wait_for(
                        func(*args, **kwargs),
                        timeout=self.config.request_timeout
                    )
                else:
                    # 同步函数在线程池中执行
                    loop = asyncio.get_event_loop()
                    result = await asyncio.wait_for(
                        loop.run_in_executor(None, func, *args, **kwargs),
                        timeout=self.config.request_timeout
                    )
                
                # 记录成功请求
                duration = time.time() - start_time
                await self._record_request(True, duration)
                
                return result
                
            except asyncio.TimeoutError:
                # 超时处理
                duration = time.time() - start_time
                await self._record_request(False, duration, FailureType.TIMEOUT, "Request timeout")
                raise
                
            except Exception as e:
                # 异常处理
                duration = time.time() - start_time
                await self._record_request(False, duration, FailureType.EXCEPTION, str(e))
                raise
    
    async def _should_allow_request(self) -> bool:
        """判断是否应该允许请求通过"""
        current_time = time.time()
        
        if self.state == CircuitState.CLOSED:
            return True
        
        elif self.state == CircuitState.OPEN:
            # 检查是否可以进入半开状态
            if (self.last_failure_time and 
                current_time - self.last_failure_time >= self.config.recovery_timeout):
                await self._transition_to_half_open()
                return True
            return False
        
        elif self.state == CircuitState.HALF_OPEN:
            # 半开状态允许部分请求
            return True
        
        return False
    
    async def _record_request(self, 
                             success: bool,
                             duration: float,
                             failure_type: Optional[FailureType] = None,
                             error_message: Optional[str] = None) -> None:
        """记录请求结果"""
        current_time = time.time()
        
        # 创建请求记录
        record = RequestRecord(
            timestamp=current_time,
            success=success,
            duration=duration,
            failure_type=failure_type,
            error_message=error_message
        )
        
        # 添加到记录列表
        self.request_records.append(record)
        
        # 清理过期记录
        await self._cleanup_old_records(current_time)
        
        # 更新统计信息
        self.stats.total_requests += 1
        if success:
            self.stats.successful_requests += 1
            self.consecutive_successes += 1
            self.consecutive_failures = 0
        else:
            self.stats.failed_requests += 1
            self.consecutive_failures += 1
            self.consecutive_successes = 0
            
            if failure_type == FailureType.TIMEOUT:
                self.stats.timeout_requests += 1
        
        # 更新平均响应时间
        if self.request_records:
            durations = [r.duration for r in self.request_records]
            self.stats.average_response_time = statistics.mean(durations)
        
        # 检查是否需要状态转换
        await self._check_state_transition()
    
    async def _cleanup_old_records(self, current_time: float) -> None:
        """清理过期的请求记录"""
        cutoff_time = current_time - self.config.timeout_window
        
        # 保留时间窗口内的记录
        self.request_records = [
            record for record in self.request_records
            if record.timestamp >= cutoff_time
        ]
        
        # 限制记录数量
        if len(self.request_records) > self.config.max_requests:
            self.request_records = self.request_records[-self.config.max_requests:]
    
    async def _check_state_transition(self) -> None:
        """检查是否需要状态转换"""
        if self.state == CircuitState.CLOSED:
            await self._check_should_open()
        elif self.state == CircuitState.HALF_OPEN:
            await self._check_should_close()
    
    async def _check_should_open(self) -> None:
        """检查是否应该打开熔断器"""
        if len(self.request_records) < self.config.min_requests:
            return
        
        # 检查连续失败次数
        if self.consecutive_failures >= self.config.failure_threshold:
            await self._transition_to_open("连续失败次数达到阈值")
            return
        
        # 检查失败率
        recent_records = self.request_records[-self.config.min_requests:]
        if len(recent_records) >= self.config.min_requests:
            failure_count = sum(1 for r in recent_records if not r.success)
            failure_rate = failure_count / len(recent_records)
            
            if failure_rate >= self.config.failure_rate_threshold:
                await self._transition_to_open(f"失败率达到阈值: {failure_rate:.2f}")
                return
    
    async def _check_should_close(self) -> None:
        """检查是否应该关闭熔断器"""
        if self.consecutive_successes >= self.config.success_threshold:
            await self._transition_to_closed("连续成功次数达到阈值")
    
    async def _transition_to_open(self, reason: str) -> None:
        """转换到打开状态"""
        if self.state != CircuitState.OPEN:
            old_state = self.state
            self.state = CircuitState.OPEN
            self.last_failure_time = time.time()
            self.stats.circuit_opens += 1
            self.stats.last_state_change = datetime.now(timezone.utc)
            
            self.logger.warning(
                f"熔断器 {self.name} 状态转换: {old_state.value} -> {self.state.value}, 原因: {reason}"
            )
    
    async def _transition_to_half_open(self) -> None:
        """转换到半开状态"""
        if self.state != CircuitState.HALF_OPEN:
            old_state = self.state
            self.state = CircuitState.HALF_OPEN
            self.consecutive_successes = 0
            self.stats.last_state_change = datetime.now(timezone.utc)
            
            self.logger.info(
                f"熔断器 {self.name} 状态转换: {old_state.value} -> {self.state.value}"
            )
    
    async def _transition_to_closed(self, reason: str) -> None:
        """转换到关闭状态"""
        if self.state != CircuitState.CLOSED:
            old_state = self.state
            self.state = CircuitState.CLOSED
            self.consecutive_failures = 0
            self.consecutive_successes = 0
            self.last_failure_time = None
            self.stats.circuit_closes += 1
            self.stats.last_state_change = datetime.now(timezone.utc)
            
            self.logger.info(
                f"熔断器 {self.name} 状态转换: {old_state.value} -> {self.state.value}, 原因: {reason}"
            )
    
    async def _handle_fallback(self, *args, **kwargs) -> Any:
        """处理降级逻辑"""
        if self.config.enable_fallback and self.fallback_func:
            try:
                # 添加降级延迟
                if self.config.fallback_delay > 0:
                    await asyncio.sleep(self.config.fallback_delay)
                
                if asyncio.iscoroutinefunction(self.fallback_func):
                    return await self.fallback_func(*args, **kwargs)
                else:
                    return self.fallback_func(*args, **kwargs)
            except Exception as e:
                self.logger.error(f"降级函数执行失败: {e}")
                raise
        
        # 如果没有降级函数，抛出熔断器异常
        raise CircuitBreakerOpenException(f"熔断器 {self.name} 处于打开状态")
    
    async def get_stats(self) -> CircuitBreakerStats:
        """获取统计信息"""
        # 计算失败率
        if self.request_records:
            failure_count = sum(1 for r in self.request_records if not r.success)
            self.stats.failure_rate = failure_count / len(self.request_records)
        
        # 更新当前状态
        self.stats.current_state = self.state
        
        return self.stats
    
    async def reset(self) -> None:
        """重置熔断器状态"""
        async with self.lock:
            self.state = CircuitState.CLOSED
            self.consecutive_failures = 0
            self.consecutive_successes = 0
            self.last_failure_time = None
            self.request_records.clear()
            
            self.logger.info(f"熔断器 {self.name} 已重置")
    
    async def force_open(self, reason: str = "手动打开") -> None:
        """强制打开熔断器"""
        await self._transition_to_open(reason)
    
    async def force_close(self, reason: str = "手动关闭") -> None:
        """强制关闭熔断器"""
        await self._transition_to_closed(reason)


class CircuitBreakerOpenException(Exception):
    """熔断器打开异常"""
    pass


class CircuitBreakerManager:
    """熔断器管理器"""
    
    def __init__(self):
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}
        self.lock = asyncio.Lock()
        self.logger = logging.getLogger(__name__)
    
    async def get_circuit_breaker(self, 
                                 name: str,
                                 config: Optional[CircuitBreakerConfig] = None,
                                 fallback_func: Optional[Callable] = None) -> CircuitBreaker:
        """获取或创建熔断器"""
        async with self.lock:
            if name not in self.circuit_breakers:
                self.circuit_breakers[name] = CircuitBreaker(
                    name=name,
                    config=config,
                    fallback_func=fallback_func
                )
                self.logger.info(f"创建熔断器: {name}")
            
            return self.circuit_breakers[name]
    
    async def remove_circuit_breaker(self, name: str) -> bool:
        """移除熔断器"""
        async with self.lock:
            if name in self.circuit_breakers:
                del self.circuit_breakers[name]
                self.logger.info(f"移除熔断器: {name}")
                return True
            return False
    
    async def get_all_stats(self) -> Dict[str, CircuitBreakerStats]:
        """获取所有熔断器的统计信息"""
        stats = {}
        for name, cb in self.circuit_breakers.items():
            stats[name] = await cb.get_stats()
        return stats
    
    async def reset_all(self) -> None:
        """重置所有熔断器"""
        for cb in self.circuit_breakers.values():
            await cb.reset()
        self.logger.info("所有熔断器已重置")


# 全局熔断器管理器
_global_circuit_breaker_manager: Optional[CircuitBreakerManager] = None


def get_circuit_breaker_manager() -> CircuitBreakerManager:
    """获取全局熔断器管理器"""
    global _global_circuit_breaker_manager
    if _global_circuit_breaker_manager is None:
        _global_circuit_breaker_manager = CircuitBreakerManager()
    return _global_circuit_breaker_manager


# 装饰器
def circuit_breaker(name: str, 
                   config: Optional[CircuitBreakerConfig] = None,
                   fallback_func: Optional[Callable] = None):
    """熔断器装饰器"""
    def decorator(func):
        async def wrapper(*args, **kwargs):
            manager = get_circuit_breaker_manager()
            cb = await manager.get_circuit_breaker(name, config, fallback_func)
            return await cb.call(func, *args, **kwargs)
        return wrapper
    return decorator
